package com.pw.study.flink.exercise;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class W2 {
    public static void main(String[] args) {
        w2();
    }

    private static void w2() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(3);
        DataStreamSource<Integer> data = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9);
        //mc(data);
        //mcKey(data);
        mcSimple(data);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void mcSimple(DataStreamSource<Integer> data) {
        mcKey(data).min(0).print("min");
        mcKey(data).max(0).print("max");
        mcKey(data).minBy(0).print("minBy");
        mcKey(data).sum(0).print("sum");
        mcKey(data).reduce((a,b)->a+b).print("reduce");
    }

    private static KeyedStream<Integer, Boolean> mcKey(DataStreamSource<Integer> data) {
        KeyedStream<Integer, Boolean> key = data.keyBy(t -> t % 5 == 0);
        return key;
    }

    private static void mc(DataStreamSource<Integer> data) {
        System.out.println("================== shuffle================");
        data.shuffle().print();
        System.out.println("================== shuffle================");
        data.keyBy(t -> t % 3 == 0).sum(0).print();
    }
}
